﻿﻿// Copyright (c) Microsoft Corporation. All rights reserved. See License.txt in the project root for license information.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;

namespace Nardax.Azure.ServiceBus
{
    public class SubscriptionAdapter : SubscriptionAdapter<BrokeredMessage>, ISubscriptionAdapter
    {
        public SubscriptionAdapter(SubscriptionAdapterDescription subscriptionAdapterDescription)
            : base(subscriptionAdapterDescription)
        {
        }
    }

    public class SubscriptionAdapter<T> : ISubscriptionAdapter<T>
    {
        private readonly ISubscriptionClientFactory _subscriptionClientFactory;
        private readonly BlockingCollection<BrokeredMessage> _brokeredMessages;
        private readonly BlockingCollection<BrokeredMessage> _deadLetterBrokeredMessages;
        private bool _isInitialized;
        private readonly object _initializationLock;
        private readonly IMessageReceiver[] _subscriptionClients;
        private readonly IMessageReceiver[] _deadLetterMessageReceivers;

        public SubscriptionAdapter(SubscriptionAdapterDescription<T> subscriptionAdapterDescription)
            : this(subscriptionAdapterDescription, new SubscriptionClientFactory())
        {
        }

        public SubscriptionAdapter(SubscriptionAdapterDescription<T> subscriptionAdapterDescription, ISubscriptionClientFactory subscriptionClientFactory)
        {
            if (subscriptionAdapterDescription == null)
            {
                throw new ArgumentNullException("subscriptionAdapterDescription");
            }

            if (subscriptionClientFactory == null)
            {
                throw new ArgumentNullException("subscriptionClientFactory");
            }

            SubscriptionAdapterDescription = subscriptionAdapterDescription;
            _subscriptionClientFactory = subscriptionClientFactory;
            _brokeredMessages = new BlockingCollection<BrokeredMessage>();
            _deadLetterBrokeredMessages = new BlockingCollection<BrokeredMessage>();
            _initializationLock = new object();
            _subscriptionClients = new IMessageReceiver[SubscriptionAdapterDescription.ConnectionStrings.Length];
            _deadLetterMessageReceivers = new IMessageReceiver[SubscriptionAdapterDescription.ConnectionStrings.Length];
        }

        public void Initialize()
        {
            try
            {
                InitializeInternal();
            }
            catch (Exception ex)
            {
                var initializationException = new ServiceBusAdapterInitializationException("SubscriptionAdapter failed to initialize.", ex);
                throw initializationException;
            }
        }

        private void InitializeInternal()
        {
            if (_isInitialized)
            {
                return;
            }

            lock (_initializationLock)
            {
                if (_isInitialized)
                {
                    return;
                }

                for (var i = 0; i < SubscriptionAdapterDescription.ConnectionStrings.Length; i++)
                {
                    TryCreateSubscription(i);
                    _subscriptionClients[i] = CreateSubscriptionClient(i);
                    _deadLetterMessageReceivers[i] = CreateDeadLetterMessageReceiver(i);
                }

                _isInitialized = true;
            }
        }

        private void TryCreateSubscription(int i)
        {
            var namespaceManager = NamespaceManager.CreateFromConnectionString(SubscriptionAdapterDescription.ConnectionStrings[i]);
            if (!SubscriptionAdapterDescription.RetryPolicy.ExecuteAction(() => namespaceManager.TopicExists(SubscriptionAdapterDescription.TopicDescription.Path)))
            {
                try
                {
                    SubscriptionAdapterDescription.RetryPolicy.ExecuteAction(() => namespaceManager.CreateTopic(SubscriptionAdapterDescription.TopicDescription));
                }
                catch (MessagingEntityAlreadyExistsException)
                {
                    // I am handling this exception because it is telling me that the topic I am trying to create already
                    // exists. This is possible because other adapters may have created it just after I checked to see
                    // if it was already there.
                }
            }

            if (!SubscriptionAdapterDescription.RetryPolicy.ExecuteAction(() => namespaceManager.SubscriptionExists(SubscriptionAdapterDescription.TopicDescription.Path, SubscriptionAdapterDescription.SubscriptionDescription.Name)))
            {
                try
                {
                    if (SubscriptionAdapterDescription.RuleDescription == null)
                    {
                        SubscriptionAdapterDescription.RetryPolicy.ExecuteAction(() => namespaceManager.CreateSubscription(SubscriptionAdapterDescription.SubscriptionDescription));
                    }
                    else
                    {
                        SubscriptionAdapterDescription.RetryPolicy.ExecuteAction(() => namespaceManager.CreateSubscription(SubscriptionAdapterDescription.SubscriptionDescription,
                                                                                                                      SubscriptionAdapterDescription.RuleDescription));
                    }
                }
                catch (MessagingEntityAlreadyExistsException)
                {
                    // I am handling this exception because it is telling me that the subscription I am trying to create already
                    // exists. This is possible because other adapters may have created it just after I checked to see
                    // if it was already there.
                }
            }
        }

        private IMessageReceiver CreateSubscriptionClient(int i)
        {
            var subscriptionClient = _subscriptionClientFactory.CreateSubscriptionClient(SubscriptionAdapterDescription.ConnectionStrings[i],
                                                                                   SubscriptionAdapterDescription.TopicDescription.Path,
                                                                                   SubscriptionAdapterDescription.SubscriptionDescription.Name);

            subscriptionClient.PrefetchCount = SubscriptionAdapterDescription.PrefetchCount;
            return subscriptionClient;
        }

        private IMessageReceiver CreateDeadLetterMessageReceiver(int i)
        {
            var path = SubscriptionClient.FormatDeadLetterPath(SubscriptionAdapterDescription.TopicDescription.Path,
                                                               SubscriptionAdapterDescription.SubscriptionDescription.Name);

            var deadLetterMessageReceiver = _subscriptionClientFactory.CreateDeadLetterMessageReceiver(SubscriptionAdapterDescription.ConnectionStrings[i], path);
            deadLetterMessageReceiver.PrefetchCount = SubscriptionAdapterDescription.DeadLetterQueuePrefetchCount;
            return deadLetterMessageReceiver;
        }

        public IEnumerable<T> GetConsumingEnumerable()
        {
            Initialize();

            // I am making the assumption that there is no data in the _brokeredMessages collection
            // and that it is necessary to seed the collection with at least one message in order
            // to get the _brokeredMessages enumeration to start its cycle. In the worst case there 
            // should only be one message per subscription client in the _brokeredMessages collection and 
            // it should be assumed that those messages have timed out.
            GetMessages();

            foreach (var brokeredMessage in _brokeredMessages.GetConsumingEnumerable())
            {
                if (!_brokeredMessages.IsAddingCompleted && _brokeredMessages.Count < 1)
                {
                    GetMessages();
                }

                if (brokeredMessage == null)
                {
                    // brokered message is only null when there are no messages in the subscription
                    // that need to be processed. This is not an exceptional case and I simply need
                    // to allow the loop to continue and eventually pull down new messages.
                    continue;
                }

                var brokeredMessageContainer = SubscriptionAdapterDescription.BrokeredMessageContainerFactory.CreateBrokeredMessageContainer(brokeredMessage);

                yield return brokeredMessageContainer;
            }
        }

        public IEnumerable<T> GetDeadLetterConsumingEnumerable()
        {
            Initialize();

            // I am making the assumption that there is no data in the _brokeredMessages collection
            // and that it is necessary to seed the collection with at least one message in order
            // to get the _brokeredMessages enumeration to start its cycle. In the worst case there 
            // should only be one message per subscription client in the _brokeredMessages collection and 
            // it should be assumed that those messages have timed out.
            GetDeadLetterMessages();

            foreach (var brokeredMessage in _deadLetterBrokeredMessages.GetConsumingEnumerable())
            {
                if (!_deadLetterBrokeredMessages.IsAddingCompleted && _brokeredMessages.Count < 1)
                {
                    GetDeadLetterMessages();
                }

                if (brokeredMessage == null)
                {
                    // brokered message is only null when there are no messages in the subscription
                    // that need to be processed. This is not an exceptional case and I simply need
                    // to allow the loop to continue and eventually pull down new messages.
                    continue;
                }

                var brokeredMessageContainer = SubscriptionAdapterDescription.BrokeredMessageContainerFactory.CreateBrokeredMessageContainer(brokeredMessage);

                yield return brokeredMessageContainer;
            }
        }

        public void StopReceiving()
        {
            _brokeredMessages.CompleteAdding();
            _deadLetterBrokeredMessages.CompleteAdding();
        }

        public void Close()
        {
            StopReceiving();

            foreach (var deadLetterQueueClient in _deadLetterMessageReceivers)
            {
                deadLetterQueueClient.Close();
            }

            foreach (var subscriptionClient in _subscriptionClients)
            {
                subscriptionClient.Close();
            }
        }

        /// <summary>
        /// all operation must result in a message being added to the _brokeredMessages collection
        /// even if that operation results in an exception... especially when it results in an exception
        /// if each call to GetMessages doesn't add something into the _BrokeredMessages collection
        /// the loop on the consuming enumerable will not exit
        /// </summary>
        private void GetMessages()
        {
            for (var i = 0; i < _subscriptionClients.Length; i++)
            {
                var sc = _subscriptionClients[i];
                try
                {
                    sc.BeginReceive(SubscriptionAdapterDescription.ServerWaitTime, ar =>
                    {
                        try
                        {
                            // if the call to EndReceive times out it will return null and not thrown an exception
                            var message = sc.EndReceive(ar);

                            sc.IsEmpty = (message == null);
                            if (sc.IsEmpty)
                            {
                                OnSubscriptionIsEmpty();
                            }

                            AddMessageToCollection(_brokeredMessages, message);
                        }
                        catch (Exception ex)
                        {
                            AddMessageToCollection(_brokeredMessages, null);
                            OnReceiveFailed(ex);
                        }
                    }, null);
                }
                catch (Exception ex)
                {
                    if (ex is OperationCanceledException)
                    {
                        // Client entity has been closed or aborted so I am going to replace it
                        _subscriptionClients[i] = CreateSubscriptionClient(i);
                    }

                    AddMessageToCollection(_brokeredMessages, null);
                    OnReceiveFailed(ex);
                }
            }
        }

        private void GetDeadLetterMessages()
        {
            for (var i = 0; i < _deadLetterMessageReceivers.Length; i++)
            {
                var dlmr = _deadLetterMessageReceivers[i];
                try
                {
                    dlmr.BeginReceive(SubscriptionAdapterDescription.ServerWaitTime, ar =>
                    {
                        try
                        {
                            // if the call to EndReceive times out it will return null and not thrown an exception
                            var message = dlmr.EndReceive(ar);

                            dlmr.IsEmpty = (message == null);
                            if (dlmr.IsEmpty)
                            {
                                OnDeadLetterQueueIsEmpty();
                            }

                            AddMessageToCollection(_deadLetterBrokeredMessages, message);
                        }
                        catch (Exception ex)
                        {
                            AddMessageToCollection(_deadLetterBrokeredMessages, null);
                            OnReceiveFailed(ex);
                        }
                    }, null);
                }
                catch (Exception ex)
                {
                    if (ex is OperationCanceledException)
                    {
                        // Client entity has been closed or aborted so I am going to replace it
                        _subscriptionClients[i] = CreateSubscriptionClient(i);
                    }

                    AddMessageToCollection(_deadLetterBrokeredMessages, null);
                    OnReceiveFailed(ex);
                }
            }
        }

        private void AddMessageToCollection(BlockingCollection<BrokeredMessage> collection, BrokeredMessage brokeredMessage)
        {
            if (!collection.IsAddingCompleted)
            {
                collection.Add(brokeredMessage);
            }
        }

        private void OnReceiveFailed(Exception exception)
        {
            if (ReceiveFailed != null)
            {
                ReceiveFailed(this, new ReceiveFailedEventArgs(exception));
            }
        }

        public event EventHandler<ReceiveFailedEventArgs> ReceiveFailed;

        private void OnSubscriptionIsEmpty()
        {
            if (SubscriptionIsEmpty != null && _subscriptionClients.All(sc => sc.IsEmpty))
            {
                SubscriptionIsEmpty(this, EventArgs.Empty);
            }
        }

        public event EventHandler SubscriptionIsEmpty;

        private void OnDeadLetterQueueIsEmpty()
        {
            if (DeadLetterQueueIsEmpty != null && _deadLetterMessageReceivers.All(dlqc => dlqc.IsEmpty))
            {
                DeadLetterQueueIsEmpty(this, EventArgs.Empty);
            }
        }

        public event EventHandler DeadLetterQueueIsEmpty;

        public SubscriptionAdapterDescription<T> SubscriptionAdapterDescription { get; private set; }
    }
}